package dao

import (
	"context"
	"github.com/nsqio/go-nsq"
	"go.uber.org/zap"
	"wxpay/internal/conf"
	"wxpay/pkg/mcontext"
	"wxpay/pkg/util"
)

type nsqLogger struct{}

// Output impl
func (n *nsqLogger) Output(_ int, s string) (err error) {
	ctx := context.Background()
	log := mcontext.Logger(ctx)
	if !util.Contains(s, "TOPIC_NOT_FOUND") {
		log.Info(s)
	}
	return
}

// 增加一个worker
func (d *WxpayDao) StartWorker(topic string, handler nsq.Handler, channel string) {
	ctx := context.Background()
	log := mcontext.Logger(ctx)
	consumer, err := nsq.NewConsumer(topic, channel, conf.Conf.NSQ)
	if err != nil {
		log.Fatal("could not create instance of Consumer with topic:"+topic, zap.Error(err))
		return
	}
	consumer.SetLogger(&nsqLogger{}, nsq.LogLevelError)
	consumer.AddConcurrentHandlers(handler, 5)
	err = consumer.ConnectToNSQLookupd(conf.Conf.NSQAddr.LookupAddress)
	if err != nil {
		log.Fatal("could not connect to nsq lookup addr", zap.Error(err))
		return
	}
	// saving for close
	d.consumers = append(d.consumers, consumer)
}
